package demo;

import java.util.Map;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

public class SentenceSpout extends BaseRichSpout {

	private static final long serialVersionUID = 1L;

	private String[] sentences = { "hello my name is jack", "i am fine thank you", "i need a cook" };

	private SpoutOutputCollector collector;

	private int index = 0;

	@Override
	public void nextTuple() {
		if (index == sentences.length) {
//			index = 0;
			Utils.waitForMillis(1);
			return;
		}
		String sentence = sentences[index++];

		collector.emit(new Values(sentence));

		Utils.waitForMillis(1);
	}

	@Override
	public void open(@SuppressWarnings("rawtypes") Map config, TopologyContext context,
			SpoutOutputCollector collector) {
		this.collector = collector;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("sentence"));
	}

}
